package com.example.yilanchatserver.business.freechat;

import com.alibaba.fastjson.JSON;
import com.example.yilanchatserver.business.freechat.enums.EnumUtils;
import com.example.yilanchatserver.business.freechat.enums.ZhiPuErrorCodes;
import lombok.extern.slf4j.Slf4j;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.Disposable;
import reactor.core.publisher.FluxSink;

import java.util.Map;

@Slf4j
public class UtopiaSubscriber implements Subscriber<String>, Disposable {
    private final FluxSink<String> emitter;
    private Subscription subscription;




    private final StringBuilder sb;

    private String history = "";


    public UtopiaSubscriber(FluxSink<String> emitter) {
        this.emitter = emitter;
        this.sb = new StringBuilder();
    }

    @Override
    public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
        subscription.request(1);
    }

    @Override
    public void onNext(String data) {
        log.info("Utopia返回数据：{}", data);
        try{
            Map paramsMap = JSON.parseObject(data, Map.class);
            String content = (String) paramsMap.get("content");
            String appUserId = (String) paramsMap.get("appUserId");
            if(StringUtils.isEmpty(content)){
                if(!StringUtils.isEmpty(appUserId)){
                    emitter.complete();
                    return;
                }
                subscription.request(1);
            } else {
                emitter.next(content);
            }

        }catch (Exception e){
            log.error(e.toString());
            emitter.next("[Error]");
            emitter.complete();
        }


    }

    @Override
    public void onError(Throwable t) {
        log.error("Utopia返回数据异常：{}", t.getMessage());
        emitter.error(t);
    }

    @Override
    public void onComplete() {
        log.info("Utopia返回数据完成");
        emitter.next("[Done]");
        emitter.complete();
    }

    @Override
    public void dispose() {
        log.warn("Utopia返回数据关闭");

        emitter.complete();
    }

}
